package study.bigdata.flink;

import cn.hutool.system.SystemUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.Map;
import java.util.Objects;

@Slf4j
public class FlinkParamSetting {

    public static void init(StreamExecutionEnvironment env, String[] args) {
        //
        ParameterTool tool = init(args);
        //
        ExecutionConfig config = env.getConfig();
        config.setGlobalJobParameters(tool);
    }

    private static ParameterTool init(String[] args) {
        args = addAnnoExt(args);
        log.info("get flink parameter start ... args: {}", Arrays.toString(args));
        //
        ParameterTool parameter = ParameterTool.fromArgs(args);

        //
        init(parameter, SPRING_CLOUD_NACOS_CONFIG_SERVER_ADDR_ALIAS, SPRING_CLOUD_NACOS_CONFIG_SERVER_ADDR, ParameterConstant.NACOS_SERVER_ADDR_MAP);
        init(parameter, SPRING_CLOUD_CONFIG_PROFILE_ALIAS, SPRING_CLOUD_CONFIG_PROFILE);
        init(parameter, JOB_NAME);
        init(parameter, JOB_NEED_INIT_DB_NAME);
        init(parameter, JOB_NEED_INIT_JEDIS);
        log.info("get flink parameter end ...");
        return parameter;
    }

    private static void init(ParameterTool parameter, String propertyNameAlias, String propertyName, Map<String, String> map) {
        String property = System.getProperty(propertyNameAlias);
        if (Objects.isNull(property)) {
            property = parameter.getRequired(propertyNameAlias);
            String val = map.get(property);
            if (Objects.isNull(val)) {
                throw new SealFlinkException(String.format("Error! Unrecognized parameter: %s = %s", propertyNameAlias, property));
            }
            System.setProperty(propertyName, val);
        }
        log.info("propertyNameAlias: {}, propertyName: {}, property: {}", propertyNameAlias, propertyName, property);
    }
}
